package com.lvmama.java.rhino.spark.utils;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.Stat;

import com.lvmama.java.rhino.spark.core.ICallback;
import com.lvmama.java.rhino.spark.core.zookeeper.ZNodeEvent;

import org.apache.zookeeper.ZooDefs.Ids;

/**
 * Zookeeper操作工具类，主要用于管理Zookeeper节点
 * @author wxliyong
 */
public class ZookeeperUtils {
	private static final Logger LOGGER = Logger.getLogger(ZookeeperUtils.class);
	private static String zookeeperUrl = Constants.getInstance().getValue("zookeeper.url");
//	private static String zookeeperUrl = "10.112.5.110:2181,10.112.5.220:2181,10.112.5.222:2181";
	private ZooKeeper zk;
	private static final int SESSION_TIMEOUT = 10000;
	private static ZookeeperUtils instance;
	private ZookeeperUtils() {
		init();
	}
	private void init() {
		try {
			CountDownLatch connectedLatch = new CountDownLatch(1);  
            Watcher watcher = new ConnectedWatcher(connectedLatch);
			zk = new ZooKeeper(zookeeperUrl, SESSION_TIMEOUT, watcher);
			waitUntilConnected(zk, connectedLatch); 
		} catch (IOException e) {
			LOGGER.error(e.getMessage(), e);
		}
	}

	private static void waitUntilConnected(ZooKeeper zk, CountDownLatch connectedLatch) {
		if (States.CONNECTING == zk.getState()) {
			try {
				connectedLatch.await();
			} catch (InterruptedException e) {
				throw new IllegalStateException(e);
			}
		}
	}
	
	static class ConnectedWatcher implements Watcher {  
		   
        private CountDownLatch connectedLatch;  
   
        ConnectedWatcher(CountDownLatch connectedLatch) {  
            this.connectedLatch = connectedLatch;  
        }  
   
        @Override  
        public void process(WatchedEvent event) {  
           if (event.getState() == KeeperState.SyncConnected) {
               connectedLatch.countDown();  
               String str = "Zookeeper service: " + zookeeperUrl + " 连接成功！";
        	   LOGGER.info(str);
        	   System.out.println(str);
           }  
        }  
    }
	
	public static ZookeeperUtils getInstance() {
		if (instance == null) {
			synchronized (ZookeeperUtils.class) {
				if (instance == null) {
					instance = new ZookeeperUtils();
				}
			}
		}
		return instance;
	}
	
	public boolean create(String path, String data) {
		try {
			this.zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			return true;
		} catch (KeeperException e) {
			LOGGER.error(e.getMessage(), e);
		} catch (InterruptedException e) {
			LOGGER.error(e.getMessage(), e);
		}
		return false;
	}

	/**
	 * 读取指定节点数据内容
	 * @param path 节点path
	 * @return
	 */
	public String read(String path) {
		try {
			return new String(this.zk.getData(path, false, null));
		} catch (KeeperException e) {
			LOGGER.error(e.getMessage(), e);
		} catch (InterruptedException e) {
			LOGGER.error(e.getMessage(), e);
		}
		return "";
	}

	/**
	 * 更新指定节点数据内容
	 * @param path 节点path
	 * @param data 数据内容
	 * @return
	 */
	public boolean write(String path, String data) {
		try {
			Stat stat = zk.setData(path, data.getBytes(), -1);
			LOGGER.info(stat.toString());
			return true;
		} catch (KeeperException e) {
			LOGGER.error(e.getMessage(), e);
		} catch (InterruptedException e) {
			LOGGER.error(e.getMessage(), e);
		}
		return false;
	}

	/**
	 * 删除指定节点
	 * @param path 节点path
	 */
	public void delete(String path) {
		try {
			this.zk.delete(path, -1);
		} catch (KeeperException e) {
			LOGGER.error(e.getMessage(), e);
		} catch (InterruptedException e) {
			LOGGER.error(e.getMessage(), e);
		}
	}
	
	public boolean exist(String path) {
		try {
			 Stat stat = this.zk.exists(path, false);
			 if(stat != null) {
				 return true;
			 }
		} catch (KeeperException e) {
			LOGGER.error(e.getMessage(), e);
		} catch (InterruptedException e) {
			LOGGER.error(e.getMessage(), e);
		}
		return false; 
	}
	
	public void registerWatcher(final String znode, final ICallback callback) {
		try {
			this.zk.exists(znode, new Watcher() {
				@Override
				public void process(WatchedEvent event) {
					ZNodeEvent zNodeEvent = new ZNodeEvent();
					zNodeEvent.setEventPath(event.getPath());
					zNodeEvent.setEventType(event.getType().toString());
					if (event.getType() == Event.EventType.None) {
						switch (event.getState()) {
						case SyncConnected:
							break;
						case Expired:
							// It's all over
							zNodeEvent.setStatus(ZNodeEvent.EventStatus.EXPIRED);
							callback.call(zNodeEvent);
							break;
						default:
							return;
						}
					}
					String path = event.getPath();
					if (path != null && path.equals(znode)) {
						callback.call(zNodeEvent);
					}
				}
			});
		} catch (KeeperException e) {
			LOGGER.error(e.getMessage(), e);
		} catch (InterruptedException e) {
			LOGGER.error(e.getMessage(), e);
		}

	}

	/**
	 * 关闭ZK连接
	 */
	public void close() {
		if (this.zk != null) {
			try {
				this.zk.close();
				instance = null;
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
}
